package com.test.rocketmq.producer;

import com.alibaba.fastjson.JSON;
import com.test.rocketmq.pojo.User;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.List;

@RestController
public class ProducerController {

    @Autowired
    private DefaultMQProducer defaultMQProducer;

    @Autowired
    private TransactionMQProducer transactionMQProducer;
    /**
     * 发送普通消息
     * */
    @GetMapping("/sendMsg")
    public void sendMsg(){
        for(int i=0;i<2;i++){
            User user = new User();
            user.setId(String.valueOf(i));
            user.setUsername("demo+"+i);
            String json = JSON.toJSONString(user);
            Message msg = new Message("user-topic","white",json.getBytes());
            try{
                SendResult result = defaultMQProducer.send(msg);
                System.out.println("消息id:"+result.getMsgId()+":"+","+"发送状态:"+result.getSendStatus());
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
    /**
     * 发送事务消息
     * */
    @GetMapping("/sendTransactionMsg")
    public void sendTransactionMsg(){
        SendResult sendResult = null;
        try{
            //a、b、c 三个值对应三个不同的状态
            String temp = "c";
            Message msg = new Message("user-topic","white",temp.getBytes());
            //发送事务消息
            sendResult = transactionMQProducer.sendMessageInTransaction(msg,(Message msg1,Object arg)->{
                String value = "";
                if (arg instanceof String) {
                    value = (String) arg;
                }
                if (value == "") {
                    throw new RuntimeException("发送消息不能为空...");
                } else if (value =="a") {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                } else if (value =="b") {
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.ROLLBACK_MESSAGE;
            },4);
            System.out.println("事务消息："+sendResult);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * 支持顺序发送消息
     */
    @GetMapping("/sendMsgOrder")
    public void sendMsgOrder(){
        for(int i=0;i<10;i++){
            User user = new User();
            user.setId(String.valueOf(i));
            user.setUsername("demo+"+i);
            String json = JSON.toJSONString(user);
            Message msg = new Message("user-topic","white",json.getBytes());
            try{
                defaultMQProducer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                        int index = ((Integer) o) % list.size();
                        return list.get(index);
                    }
                },i);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}
